import amqp from 'amqplib'
import config from '../config/rabbitmq.js';
import dayjs from 'dayjs'

const queue_send = 'send_queue';    //发送队列
const queue_retry = 'send_queue_retry'    //重试队列
const queue_fail = 'send_queue_fail'      //失败队列 ，给人工查看错误
const dlx_routing_exchange = 'dlx_routing_exchange'
const dlx_routing_key = 'dlx_routing_key'
// 声明一个延时消息交换机和一个队列
const delayedExchangeName = 'delayed_exchange';
const delayedQueueName = 'delayed_queue';

class RabbitConsumer {
  constructor() {
    //this.connect()
  }

  async connect () {
    console.log('connect');
    try {
      this.connection = await amqp.connect(`amqp://${config.username}:${config.password}@${config.hostname}:${config.port}`);

      this.channel = await this.connection.createChannel()

      await this.channel.assertQueue(queue_send, { durable: true })
      //创建死信交换机
      await this.channel.assertExchange(dlx_routing_exchange)
      //绑定死信路由,绑定死信交换机
      await this.channel.bindQueue(queue_send, dlx_routing_exchange, dlx_routing_key)


      await this.channel.assertQueue(queue_retry,
        {
          durable: true,
          deadLetterExchange: dlx_routing_exchange,
          deadLetterRoutingKey: dlx_routing_key,
          messageTtl: 30 * 1000   //重试时间
        })
      //失败后
      await this.channel.assertQueue(queue_fail, { durable: true })

      await this.channel.assertExchange(delayedExchangeName, 'x-delayed-message', {
        arguments: { 'x-delayed-type': 'direct' }
      });
      await this.channel.assertQueue(delayedQueueName, { durable: true });

      // 将队列绑定到交换机上，并设置延时时间（例如10秒）
      await this.channel.bindQueue(delayedQueueName, delayedExchangeName, '', {
        'x-delay': 10000
      });
    }
    catch (err) {
      console.log('err', err);
    }
  }
  //发送队列 (channel是否要一直关闭)
  async sendToWorker (queue, msg, count) {
    try {

      await this.channel.sendToQueue(queue, Buffer.from(msg), {
        deliveryMode: true,//表示消息需要持久化存储。
        //传递信息
        headers: {
          count
        }
      })
    }
    catch (err) {
      console.log('err', err);
    }
  }

  sendEmail (msg) {
    console.log('type:', typeof msg, JSON.parse(msg.toString()));
    return Math.random() * 100 > 50 ? true : false
  }

  recevie (msg, count) {
    if (count >= 3) {
      return true;
    }
    else {
      console.log('type:', typeof msg, JSON.parse(msg.toString()));
      return Math.random() * 100 > 50 ? true : false
    }

  }

  //消费队列
  async consumerByWorker () {
    await this.channel.prefetch(1) //设置通道的预获取计数为 1，确保一次只处理一条消息。
    await this.channel.consume(queue_send, (msg) => {

      let retryCount = msg.properties.headers.count
      const result = this.sendEmail(msg.content);
      console.log(`${dayjs().format('YYYY-MM-DD HH:mm:ss')}--->接受数据处理` + result);
      if (result) { //发送成功
        console.log('发送成功')
        this.channel.ack(msg);
      }
      else if (!result && retryCount < 3) { //发送失败重试，且重试次数小于3
        console.log(`处理数据，第${retryCount}次重试`)
        this.sendToWorker(queue_retry, msg.content.toString(), ++retryCount)
        //重新投递
        this.channel.ack(msg); //nack自动投递
      } else {
        console.log(`重试${retryCount}次后仍然失败，不再处理`)
        this.sendToWorker(queue_fail, msg.content.toString(), retryCount)
        //不管是否发送成功能都要ack
        this.channel.ack(msg)
      }

    }, {
      noAck: false //是否自启动确认消费
    })
  }

  //消费队列
  async consumerByWorkerLater () {
    await this.channel.prefetch(1) //设置通道的预获取计数为 1，确保一次只处理一条消息。
    await this.channel.consume(delayedQueueName, (msg) => {

      let retryCount = msg.properties.headers.count
      const result = this.recevie(msg.content, retryCount);
      console.log(`later ${dayjs().format('YYYY-MM-DD HH:mm:ss')}--->接受数据处理` + result);
      if (result) { //发送成功
        console.log('later 发送成功')
        this.channel.ack(msg);
      }
      else { //发送失败重试，且重试次数小于3
        msg.properties.headers.count++;
        console.log(`处理数据，第${retryCount}次重试`)
        //重新投递
        this.channel.nack(msg); //nack自动投递
      }

    }, {
      noAck: false //是否自启动确认消费
    })
  }
}

var instance = new RabbitConsumer();

export default instance;